package com.pxene.dmp.reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.pxene.dmp.common.KVSender;
import com.pxene.dmp.common.KVUrlGenerator;
import com.pxene.dmp.constant.BaseConstant;

public class CountDataSizeReducer extends
		Reducer<Text, LongWritable, NullWritable, NullWritable> {
	
	private Log log = LogFactory.getLog(CountDataSizeReducer.class);
	
    private String date;
    private String apiName;
    
    private static final String DEFAULT_PROVINCE = "BJ";
    private static final String DEFAULT_IMSI = "ABC1DEF2GHI3GKL4MNO5PQR6STU7VWX8";
    private static final String DEFAULT_MEID = "ABC1DEF2GHI3GKL4MNO5PQR6STU7VWX8";
    private static final String DEFAULT_DOMAIN = "00000000";
    private static final String DEFAULT_TIME = "0123456789";
    private static final String DEFAULT_DATA_TYPE = "0";
    
    private Map<String, String> params = new HashMap<String, String>();
    
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
		date = conf.get("date");
		apiName = conf.get("api-name");
		log.info(BaseConstant.LOG_PREFIX + "date=" + date + ",apiName=" + apiName);
	}

	@Override
	protected void reduce(Text key, Iterable<LongWritable> values, Context context)
			throws IOException, InterruptedException {
		Iterator<LongWritable> iter = values.iterator();
		long maxIndex = 0;
		while (iter.hasNext()) {
			long index = iter.next().get();
			if (index > maxIndex) {
				maxIndex = index;
			}
		}
		
		String url = null;
		try {
			url = KVUrlGenerator.getCustUrl(apiName);
		} catch (Exception e) {
			e.printStackTrace();
		}
		params.clear();
		switch (apiName) {
		case BaseConstant.API_SETBUSINESSCODE:
			// code是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("imsi", DEFAULT_IMSI);
			params.put("meid", DEFAULT_MEID);
			params.put("name", "JD");
			params.put("code", String.valueOf(maxIndex));
			params.put("time", DEFAULT_TIME);
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_SETDOMAINUSER:
			// count是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("domain", DEFAULT_DOMAIN);
			params.put("imsi", DEFAULT_IMSI);
			params.put("count", String.valueOf(maxIndex));
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_SETSEARCHWORD:
			// word是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("imsi", DEFAULT_IMSI);
			params.put("meid", DEFAULT_MEID);
			params.put("word", String.valueOf(maxIndex));
			params.put("time", DEFAULT_TIME);
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_SETUSERINFO:
			// count是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("adid", "abcdefghijklmnop");
			params.put("type", "3");
			params.put("imsi", DEFAULT_IMSI);
			params.put("meid", DEFAULT_MEID);
			params.put("useragent", "other");
			params.put("source", "baidu.com");
			params.put("count", String.valueOf(maxIndex));
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_SETWEIXINBIZ:
			// biz是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("imsi", DEFAULT_IMSI);
			params.put("meid", DEFAULT_MEID);
			params.put("biz", String.valueOf(maxIndex));
			params.put("time", DEFAULT_TIME);
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_WEATHERCITY:
			// city是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("imsi", DEFAULT_IMSI);
			params.put("meid", DEFAULT_MEID);
			params.put("domain", DEFAULT_DOMAIN);
			params.put("city", String.valueOf(maxIndex));
			params.put("time", DEFAULT_TIME);
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_AUTOINFO:
			// auto是数据长度
			params.put("index", "0");
			params.put("date", date);
			params.put("imsi", DEFAULT_IMSI);
			params.put("meid", DEFAULT_MEID);
			params.put("domain", DEFAULT_DOMAIN);
			params.put("city", "c-00000000");
			params.put("auto", String.valueOf(maxIndex));
			params.put("time", DEFAULT_TIME);
			params.put("province", DEFAULT_PROVINCE);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_DOMAININFO:
			params.put("index", "0");
			params.put("date", date);
			params.put("province", DEFAULT_PROVINCE);
			params.put("imsi", DEFAULT_IMSI);
			params.put("domain", DEFAULT_DOMAIN);
			params.put("adid", String.valueOf(maxIndex));
			params.put("time", DEFAULT_TIME);
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		case BaseConstant.API_FAVOURITESITE:
			params.put("index", "0");
			params.put("date", date);
			params.put("province", DEFAULT_PROVINCE);
			params.put("imsi", DEFAULT_IMSI);
			params.put("site", String.valueOf(maxIndex));
			params.put("datatype", DEFAULT_DATA_TYPE);
			break;
		default:
			break;
		}
		CountDownLatch latch = new CountDownLatch(1);
		// 异步调用
		KVSender sender = new KVSender(url, params, latch);
		sender.execute();
		// 无论是否成功，都向下进行
		latch.await();
		sender.close();
		
		context.write(NullWritable.get(), NullWritable.get());
	}
	
}
